package dstbsort;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

import sort.RadixSort;
import utils.FileUtils;
import utils.SortUtils;

//分布式基数排序
public class DstbRadixSort {
    public static void main(String[] args) throws Exception {
        // 需要加载的排序的文件，n-普通版本，数据范围在int范围内，1k - 1000个数据的文件
//        String fileName = "n_1k";
//		String fileName = "n_1w";
//		String fileName = "n_10w";
//        String fileName = "n_100w";
		String fileName = "n_500w";

        // 数据文件加载
        long startTime = System.currentTimeMillis();
        String readText = FileUtils.readText(SortUtils.FILE_PREFIX + fileName);
        String[] numString = readText.split(" ");
        long endTime = System.currentTimeMillis();
        System.out.println(fileName + "数据文件加载共耗时：" + (endTime - startTime) + "ms");

        // 用多线程实现基数排序
        startTime = System.currentTimeMillis();
        int[] num = dstbRadixSort(numString);
        endTime = System.currentTimeMillis();
        System.out.println(fileName + "多线程基数排序共耗时：" + (endTime - startTime) + "ms");

        // 正确性验证
        startTime = System.currentTimeMillis();
        SortUtils.validate(num);
        endTime = System.currentTimeMillis();
        System.out.println(fileName + "基数排序正确性验证共耗时：" + (endTime - startTime) + "ms");
    }

    private static int[] dstbRadixSort(String[] numString) throws Exception {
        // 数据按位数分组，每组用一个线程来处理
        // int类型数据，最大有10位
        int n = 10;
        // 初始化buckets，共10个，按位数分到不同的list中
        ArrayList<ArrayList<Long>> buckets = new ArrayList<ArrayList<Long>>(n) {
            {
                for (int i = 0; i < n; i++) {
                    add(new ArrayList<>());
                }
            }
        };
        long[] maxNum = new long[n];
        // 记录排序数字总个数
        int total = 0;
        for (String value : numString) {
            long sortValue = Long.valueOf(value) + Integer.MAX_VALUE;
            int bit = ((sortValue + "").length() - 1); // 计算这个数有多少位、... bit表示存到第几桶
            buckets.get(bit).add(sortValue);
            if (maxNum[bit] < sortValue) {
                maxNum[bit] = sortValue;
            }
            total++;
        }
        // 释放内存
        numString = null;


        ExecutorService threadPool = Executors.newFixedThreadPool(n);
        List<Future<int[]>> futureList = new ArrayList<>(n);
        for (int i = 0; i < n; i++) {
            if (buckets.get(i).size() > 0) {
                Long[] value = new Long[buckets.get(i).size()];
                futureList.add(threadPool.submit(new MyTask(buckets.get(i).toArray(value), maxNum[i])));
            }
        }
        // 将多线程排序结果合并起来
        int[] num = new int[total];
        int start = 0;
        for (int i = 0; i < futureList.size(); i++) {
            int[] result = futureList.get(i).get();
            System.arraycopy(result, 0, num, start, result.length);
            start += result.length;
        }
        return num;
    }

    public static class MyTask implements Callable<int[]> {
        Long[] num;
        Long max;

        public MyTask(Long[] num, Long max) {
            this.num = num;
            this.max = max;
        }

        @Override
        public int[] call() throws Exception {
            return RadixSort.sort(num, max);
        }
    }
}
